/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.lidejia.hivetotalsort.type;

import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import org.apache.hadoop.io.GenericWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

/**
 *
 * @author lidejia
 */
public class HiveType extends GenericWritable implements WritableComparable<HiveType> {

    static {
        WritableComparator.define(HiveType.class, new HiveTypeComparator());
    }
    //
    public final static Class[] SUPPORTED_CLASSES = {IntWritable.class, LongWritable.class, Text.class};

    @Override
    protected Class<? extends Writable>[] getTypes() {
        return SUPPORTED_CLASSES;
    }
    //
    private boolean orderReverse = false;

    public HiveType() {
        super();
    }

    public HiveType(WritableComparable value, boolean reverse) {
        super();
        this.set(value);
        this.orderReverse = reverse;
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        int length = in.readInt();
        orderReverse = in.readBoolean();
        super.readFields(in);
    }

    @Override
    public void write(DataOutput out) throws IOException {
        ByteArrayOutputStream bufferStream = new ByteArrayOutputStream();
        DataOutput bufferDataOut = new DataOutputStream(bufferStream);
        super.write(bufferDataOut);
        byte[] buffer = bufferStream.toByteArray();
        //
        int length = buffer.length + 1 + 4; // data + orderreverse + length
        out.writeInt(length);
        //
        out.writeBoolean(orderReverse);
        out.write(buffer);
    }

    public void setValue(WritableComparable value) {
        this.set(value);
    }

    public int compareTo(HiveType o) {
        WritableComparable self = (WritableComparable) this.get();
        WritableComparable that = (WritableComparable) o.get();
        int cmp = self.compareTo(that);
        if (orderReverse) {
            cmp = -cmp;
        }
        return cmp;
    }
}
